package org.yaac.server.egql.processor;

import java.util.List;

import org.yaac.shared.egql.ResultStatus;

import com.google.appengine.tools.pipeline.Job2;
import com.google.appengine.tools.pipeline.Value;

/**
 * @author Max Zhu (thebbsky@gmail.com)
 *
 */
public class ProcessorManager extends Job2<Void, List<Processor>, ProcessContext> {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	/**
	 * task queue's deadline is 10 mins, here we make it 9 min (540)
	 */
	public static final long LIMIT = 540 * 1000;
	
	boolean stillHaveTimeForTheNextBatch(int processedBatch, long start, long now) {		
		long timeElapsed = now - start;
		long timePerBatch = timeElapsed / processedBatch;
		
		return timeElapsed + timePerBatch < LIMIT;
	}

	@Override
	public Value<Void> run(List<Processor> processors, ProcessContext context) {
		// init context with current pipeline key, as this can't be obtained outside
		context.setPipelineKey(this.getJobKey());
		
		// start process
		long start = System.currentTimeMillis();
		int processedBatch = 0;
		
		while (true) {
			ProcessData param = null;
			for (Processor processor : processors) {
				param = processor.process(context, param);
			}
			
			if (context.getStatus() == ResultStatus.PROCESSING) {
				// more entities to process
				long now = System.currentTimeMillis();

				if (!stillHaveTimeForTheNextBatch(++processedBatch, start, now)) {
					return futureCall(this, immediate(processors), immediate(context));
				} 
			}  else {
				// job finished, simply return
				return null;
			}
		}
	}
}
